package com.zhoulz.homework.chaptor02;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CreateBuilder;
import org.apache.curator.framework.api.DeleteBuilder;
import org.apache.curator.framework.api.ExistsBuilder;
import org.apache.curator.utils.CloseableUtils;
import org.apache.log4j.Logger;

import com.zhoulz.homework.chaptor02.config.Config;
import com.zhoulz.homework.chaptor02.zk.Zk;

/**
 * <p>1、init 程序,初始化zookeeper配置目录</p>
 * <pre>
 * 	app
 * 	|-config
 * 	|	nodes
 * 	|-nodes
 * 	|-commands
 * </pre>
 * <p>2、生成100个10k~10M随机数据文件到data目录下：1.txt ~ 100.txt</p>
 * @author zhoulz
 *
 */
public class Init {
	static Logger log = Logger.getLogger(Init.class);
	public static void main(String[] args) throws Exception {
		// zookeeper初始化程序
		CuratorFramework client = null;
		try {
			log.info("开始初始化zk配置目录。");
			Config.load(Config.LOCAL_CONFIG_NAME);
			client = Zk.newClient(String.format("%s:%s", 
					Config.get(Config.KEY_ZK_SERVER_IP), 
					Config.get(Config.KEY_ZK_SERVER_PORT)));
			client.start();
			if(client.getZookeeperClient().blockUntilConnectedOrTimedOut()){
				ExistsBuilder existsBuild = client.checkExists();
				DeleteBuilder delBuild = client.delete();
				CreateBuilder createBuild = client.create();
				// 先删除应用目录
				if(existsBuild.forPath(Zk.APP) != null){
					delBuild.deletingChildrenIfNeeded().forPath(Zk.APP);
				}
				createBuild.creatingParentsIfNeeded().forPath(Zk.APP_NODE_CONFIG);
				createBuild.creatingParentsIfNeeded().forPath(Zk.APP_NODE_PATH);
				createBuild.creatingParentsIfNeeded().forPath(Zk.APP_COMMANDS_PATH);
				// 写入5个节点配置，端口分别为8001,8002,8003,8004,8005
				int port = 8000;
				for (int i = 1; i <= 5; i++) {
					StringBuilder sb = new StringBuilder();
					sb.append("tcp.host.ip=127.0.0.1").append("\r\n")
					.append("tcp.host.port=").append(port + i).append("\r\n")
					.append("tcp.keepalive=1");
					
					String path = String.format("%s/%s-%d", Zk.APP_NODE_CONFIG, "leader", i);
					createBuild.forPath(path, sb.toString().getBytes());
				}
				log.info("zookeeper配置初始化完成。");
				
				log.info(String.format("开始在目录[%s]下生成随机的数据文件!", Config.DATA_DIR));
				
				generateDataFiles();
				
				log.info("初始化完成, 5秒后退出！");
			}else{
				log.info("zookeeper配置初始化失败，请先修改config/zookeeper.properties文件下的iP和port");
			}
			
		} catch (Exception e) {
			log.error("zookeeper初始化异常", e);
		}finally{
			CloseableUtils.closeQuietly(client);
		}
		for (int i = 5; i > 0; i--) {
			System.out.println(i);
			Thread.sleep(1000);
		}
		System.exit(0);
	}
	
	/**
	 * 生成数据文件
	 * @throws InterruptedException
	 */
	private static void generateDataFiles() throws InterruptedException {
		Random rand = new Random(System.currentTimeMillis());
		// 10K ~ 10M
		final int min = 10 * 1024, max = 10 * 1024 * 1024;
		int size = 0;
		final byte[] msgs = "You are Best! ".getBytes();
		
		// 建一个大小10的线程池
		ExecutorService ec = Executors.newFixedThreadPool(10);
		for (int i = 1; i <= 100; i++) {
			// 10 k ~ 10M
			while((size = rand.nextInt(max)) < min){
				continue;
			}
			/*FileChannel fc = new RandomAccessFile(file,"rw").getChannel();
				MappedByteBuffer out = fc.map(FileChannel.MapMode.READ_WRITE, 0, size);
				for (int j = 0; j < size; j++) {  
		            out.put(msgs[j % msgs.length]);
		        }
				out.clear();
				fc.close();*/
			final int count = i, length = size;
			ec.submit(new Runnable() {
				
				public void run() {
					// 1.txt ~ 100.txt
					File file = new File(Config.DATA_DIR.
							concat(File.separator).
							concat(String.valueOf(count).concat(".txt")));
					if(!file.exists()){
						try {
							file.createNewFile();
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					// 将数据写入bytebuf
					ByteBuf buf = Unpooled.buffer(length);
					while(buf.writableBytes() > 0){
						if(buf.writableBytes() < msgs.length){
							buf.writeBytes(msgs, 0, buf.writableBytes());
						}else{
							buf.writeBytes(msgs);
						}
					}
					// 将bytebuf的数据读进文件流
					FileOutputStream out = null;
					try {
						out = new FileOutputStream(file);
						buf.readBytes(out, length);
					} catch (Exception e) {
						e.printStackTrace();
					}finally{
						try {
							out.flush();
							out.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
						// 释放buf
						buf.release(buf.refCnt());
						System.out.println(file.getName());
					}
					
				}
			});
		}
		// 关闭线程池，只是关闭task通道，但是已经启动的task会执行完成。
		ec.shutdown();
		// 等待所有线程完成
		ec.awaitTermination(5, TimeUnit.MINUTES);
	}

}
